package org.acme.latch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/*
 * @className: MySqlLatch
 * @author: chentao
 * @date: 2025/2/24 上午10:54
 * @Version: 1.0
 * @description:
 */
public class MySqlLatch {

    Logger logger = LoggerFactory.getLogger(getClass());

    private DataSource dataSource;

    private Connection connection;

    private PreparedStatement preparedStatement;

    private boolean connAutoCommit = false;

    //分布式锁的SQL
    private String SELECT_FOR_UPDATE_SQL = "select * from sys_lock where id = 2 for update";

    //维持锁活跃的SQL
    private String ACTIVE_SELECT_SQL = "select 1";

    private volatile boolean running = true;

    private Thread thread;

    //心跳线程池核心数
    private int corePoolSize = 1;

    //心跳线程池心跳频率 默认30秒
    private int timeInterval = 10;

    //心跳线程池心跳的单位 默认秒
    private TimeUnit timeUnit = TimeUnit.SECONDS;

    private MysqlLatchListener mysqlLatchListener;

    private ScheduledThreadPoolExecutor threadPool;

    public MySqlLatch(DataSource dataSource, MysqlLatchListener mysqlLatchListener) {
        this.dataSource = dataSource;
        this.mysqlLatchListener = mysqlLatchListener;
        this.threadPool = new ScheduledThreadPoolExecutor(this.corePoolSize);
    }

    /****
     * 开启线程并开始选举操作
     * @throws Exception
     */
    public void start() throws Exception {
        this.thread = new Thread(new Runnable() {
            @Override
            public void run() {
                boolean lockAcquired = false;
                while (running) {
                    try {
                        connection = dataSource.getConnection();
                        connAutoCommit = connection.getAutoCommit();
                        connection.setAutoCommit(false);
                        preparedStatement = connection.prepareStatement(SELECT_FOR_UPDATE_SQL);
                        preparedStatement.execute();
                        lockAcquired = true;
                        break;
                    } catch (Exception e) {
                        closeAndClean();
                    }
                }
                if (lockAcquired) {
                    threadPool.scheduleAtFixedRate(new ConnectionActiveThread(), 0, timeInterval, timeUnit);
                    mysqlLatchListener.isLeader();
                } else {
                    logger.error("Failed to acquire lock after multiple attempts.");
                }
            }
        });
        this.thread.setName("mysql-latch-thread");
        this.thread.setDaemon(true);
        this.thread.start();
    }

    /****
     * 停止并回收线程
     * @throws Exception
     */
    public void stop() {
        this.running = false;
        this.closeAndClean();
        if (this.threadPool != null && !this.threadPool.isShutdown()) {
            this.threadPool.shutdownNow();
        }
        if (this.thread != null && this.thread.isAlive()) {
            this.thread.interrupt();
        }
    }

    /****
     * 关闭并回收资源
     */
    private void closeAndClean() {
        if (connection != null) {
            try {
                connection.commit();
            } catch (SQLException e) {
                logger.warn("connection commit error: " + e.getMessage());
            }
        }
        try {
            if (connection != null) {
                connection.setAutoCommit(connAutoCommit);
            }
        } catch (SQLException e) {
            logger.warn("connection set autocommit error: " + e.getMessage());
        }
        if (connection != null) {
            try {
                connection.close();
            } catch (SQLException e) {
                logger.warn("connection close error: " + e.getMessage());
            }
        }
        if (preparedStatement != null) {
            try {
                preparedStatement.close();
            } catch (SQLException e) {
                logger.warn("preparedStatement close error: " + e.getMessage());
            }
        }
    }

    /****
     * 保持链接心跳处理类
     */
    class ConnectionActiveThread implements Runnable {

        @Override
        public void run() {
            PreparedStatement statement = null;
            try {
                statement = connection.prepareStatement(ACTIVE_SELECT_SQL);
                statement.execute();
            } catch (Exception e) {
                logger.warn("心跳线程查询出错" + e.getMessage());
            } finally {
                if (null != statement) {
                    try {
                        statement.close();
                    } catch (SQLException e) {
                        logger.warn("心跳线程statement" + e.getMessage());
                    }
                }
            }
        }
    }
}
